
#define OP_JOIN             1
#define OP_UNION            2
#define OP_INTERSECTION     3
#define OP_MINUS            4

int multiway_sort_op(int b_start_R, int b_start_S, int b_start_result, Buffer *buf, int op){
    unsigned char* blk_R = readb(b_start_R, buf);
    unsigned char* blk_S = readb(b_start_S, buf);
    unsigned char* blk_result = getNewBlockInBuffer(buf);
    int p_S = 0;
    int p_R = 0;
    int p_result = 0;
    int b_S = b_start_S;
    int b_R = b_start_R;
    int b_result = b_start_result;
    int value_R=0;
    int value_S=0;
    int end = 0; // ture if traverse all data of R or S

    clear_buffer(blk_result);
    while(1){
        value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        value_S = get_value(get_turple_addr(p_S, blk_S), 0);
        // R moves to next
        while(value_R > value_S){
            if(move_next(&p_R, &b_R, &blk_R, b_start_R + SIZE_R - 1, buf)){
                if(op == OP_MINUS){
                    freeBlockInBuffer(blk_R, buf);
                    printf("rest S");
                    return  copy_rest_relative(p_S, b_S, blk_S,
                            b_start_S+SIZE_S-1,p_result, b_result, blk_result, buf);
                }
                else {
                    freeBlockInBuffer(blk_R, buf);
                    freeBlockInBuffer(blk_S, buf);
                    write_lastblk(b_result, 0, blk_result, buf);
                    return b_result;
                }
            }
            value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        }
        if(end == 1)
            break;

        if(end == 0 && value_R == value_S){
            int p_R_tmp = p_R; 
            int b_R_tmp = b_R; 
            unsigned char *blk_R_tmp = readb(b_R_tmp, buf);
            int match = 0;
            while(end == 0 && value_R == value_S){
                if(op == OP_JOIN){
                    // every result tuple consists of double size of origin
                    if(p_result == (MAX_TUPLE_PBLK/2)*2 ){
                        write_lastblk_next(b_result++, blk_result, buf);
                        blk_result = getNewBlockInBuffer(buf);
                        clear_buffer(blk_result);
                        p_result = 0;
                    }
                    turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_R_tmp, blk_R_tmp));
                    turple_cpy(get_turple_addr(p_result+1, blk_result), get_turple_addr(p_S, blk_S));
                    p_result += 2;
                }
                else if(op == OP_INTERSECTION){
                    if(is_turple_equal(get_turple_addr(p_R_tmp, blk_R_tmp), get_turple_addr(p_S, blk_S))){
                        check_result_blk_full(&p_result, &b_result, &blk_result, buf);
                        turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_R_tmp, blk_R_tmp));
                        p_result ++;
                    }
                }
                else if(op == OP_MINUS){
                    if(is_turple_equal(get_turple_addr(p_R_tmp, blk_R_tmp), get_turple_addr(p_S, blk_S))){
                        match = 1;
                        break;
                    }
                }
                

                // R moves to next
                if(move_next(&p_R_tmp, &b_R_tmp, &blk_R_tmp, b_start_R + SIZE_R - 1, buf)){
                    freeBlockInBuffer(blk_R_tmp, buf);
                    break; 
                }
                value_R = get_value(get_turple_addr(p_R_tmp, blk_R_tmp), 0);
            }
            if(op == OP_MINUS){
                if(!match){
                    check_result_blk_full(&p_result, &b_result, &blk_result, buf);
                    turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_S, blk_S));
                    p_result ++;
                }
            }
            freeBlockInBuffer(blk_R_tmp, buf);
        }
        else{
            if(op == OP_MINUS){
                check_result_blk_full(&p_result, &b_result, &blk_result, buf);
                turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_S, blk_S));
                p_result ++;
            }
        }
        // S moves to next
        if(move_next(&p_S, &b_S, &blk_S, b_start_S + SIZE_S - 1, buf)){
            freeBlockInBuffer(blk_R, buf);
            freeBlockInBuffer(blk_S, buf);
            write_lastblk(b_result, 0, blk_result, buf);
            return b_result;
        }
    }
    return b_result;
}